package com.mjf.spark.day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * TopN案例
 */
object Spark10_TopN {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark10_TopN")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    // 读取数据文件   时间戳 省份id 城市id 用户id 广告id
    val file: RDD[String] = sc.textFile("D:\\code\\spark\\input\\agent.log")

    // 转换为(省份id-广告id, 1)的结构
    val mapRDD: RDD[(String, Int)] = file.map(
      line => {
        val fields: Array[String] = line.split(" ")
        (fields(1).concat("-").concat(fields(4)), 1)
      }
    )

    // 按照 省份id-广告id 进行聚合
    val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    // 转换为(省份id, (广告id, 次数))的结构
    val newMapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
      case (str, num) =>{
        val arr: Array[String] = str.split("-")
        (arr(0), (arr(1), num))
      }

    }

    // 按照省份进行分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()

    // 对每个省份的广告点击次数按降序排序，取前三名
    val resRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      it => it.toList.sortBy(- _._2).take(3)
    )

    resRDD.collect().foreach(println)

    // 关闭连接
    sc. stop()

  }
}
